package org.qiyu.live.id.generate.provider.service.impl;

import ch.qos.logback.classic.util.LogbackMDCAdapter;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.Resource;
import org.qiyu.live.id.generate.provider.entity.IdGeneratePO;
import org.qiyu.live.id.generate.provider.service.IdGenerateService;
import org.qiyu.live.id.generate.provider.mapper.IdGenerateMapper;
import org.qiyu.live.id.generate.provider.service.bo.LocalSeqIdBO;
import org.qiyu.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author c
 * @description 针对表【t_id_generate_config】的数据库操作Service实现
 * @createDate 2024-02-29 15:13:29
 */
@Service
public class IdGenerateServiceImpl extends ServiceImpl<IdGenerateMapper, IdGeneratePO>
        implements IdGenerateService, InitializingBean {

    public static final int SEQ_ID = 1; //id 有序 / 无序
    @Resource
    private IdGenerateMapper idGenerateMapper;

    private static final Logger LOGGER = LoggerFactory.getLogger(IdGenerateServiceImpl.class);

    /**
     *     分布式id生成策略 涉及到多个线程的并发访问
     *
     *     HashMap不是线程安全的，即多线程下，HashMap的put和get操作可能会出错
     *
     *    ConcurrentHashMap是线程安全
     *
     *
     *    localSeqIdBOMap如何初始化，
     *    IdGenerateServiceImpl由spring容器管理，当前bean 初始化时，可以通过实现 InitializingBean 接口
     *    在 @Override
     *     public void afterPropertiesSet() throws Exception {
     *
     *     }中，也就是在bean初始化时，同时去初始化localSeqIdBOMap
     *
     *     初始化localSeqIdBOMap，需要向发号机申请号段
     *
     *     号段快用完时，要提前触发一个异步任务，再次去抢占新的号段，提前触发是为了保证业务的正常运行，抢占号段是乐观锁控制的
      */

    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<>();
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<>();

    private static final float UPDATE_RATE = 0.75F;

    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();


    //定义线程池，其中线程数量不是一个固定的值，而是一个范围  8- 16  表明最少都有8个线程存活
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1000),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("id-generate-thread-" + ThreadLocalRandom.current().nextInt(1000));
                    return thread;
                }
            });


    /**
     * 获取本地有序id
     * @param id
     * @return
     */
    @Override
    public Long getSeqId(Integer id) {
        if (id == null) {
            LOGGER.error("[getSeqId] id is error,id is {}", id);
            return null;
        }
//        id 对应 分布式id生成策略
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);  //取出相应的策略模式
        if (localSeqIdBO == null) {
            LOGGER.error("[getSeqId] localSeqIdBO is error,id is {}", id);
            return null;
        }

        // 获取生成的id ( currentNum) 多个线程并发访问，并需要进行 +1 操作
        // 当前id 实际上不能一直做自增，在达到阈值的时候，需要重新想发号机申请新的号段
        //判断是否要去预先申请新的号段
        this.refreshLocalSeqId(localSeqIdBO);
        long returnId = localSeqIdBO.getCurrentNum().getAndIncrement()  ;

        if (returnId >= localSeqIdBO.getNextThreshold()) {
            LOGGER.error("[getSeqId] id is over limit,id is {}", id);
            return null;
        }


        return returnId;
    }

    /**
     * 获取本地无序id
     * @param id
     * @return
     */
    @Override
    public Long getUnSeqId(Integer id) {

        //获取无序ID
        if (id == null) {
            LOGGER.error("[getSeqId] id is error,id is {}", id);
            return null;
        }
        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(id);
        if (localUnSeqIdBO == null) {
            LOGGER.error("[getSeqId] localSeqIdBO is error,id is {}", id);
            return null;
        }
        Long returnId = localUnSeqIdBO.getIdQueue().poll();

        if (returnId == null) {
            LOGGER.error("[getUnSeqId] localUnSeqIdBO is null,id is {}", id);
            return null;
        }
        this.refreshLocalUnSeqId(localUnSeqIdBO);


        return returnId;
    }


    /**
     * 刷新本地有序id段
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if (localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart() > step * UPDATE_RATE) {

            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null,id is {}", localSeqIdBO.getId());
                return;
            }

            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                //  这里是需要去发号机申请新的号段，如果使用同步请求，会降低性能


                LOGGER.info("开始尝试进行本地id段的同步操作");
                LOGGER.info("当前号段已经使用到"+localSeqIdBO.getCurrentNum().get());
                threadPoolExecutor.execute(()->{
                    try {
                        IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                        tryUpdateMysqlRecord(idGeneratePO);
                    }
                    catch (Exception e) {
                        LOGGER.error("同步id段失败,id is {}", localSeqIdBO.getId());
                    }finally {
                        semaphoreMap.get(localSeqIdBO.getId()).release();
                        LOGGER.info("本地有序id段同步完成,id is {}", localSeqIdBO.getId());
                    }

                });

            }
        }
    }

    /**
     * 刷新本地无序id段
     * @param localUnSeqIdBO
     */
    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        long step = localUnSeqIdBO.getNextThreshold() - localUnSeqIdBO.getCurrentStart();

        // 如果队列中的元素小于25%，则需要去申请新的号段
        if (localUnSeqIdBO.getIdQueue().size() < step * (1 - UPDATE_RATE)) {

            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null,id is {}", localUnSeqIdBO.getId());
                return;
            }

            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                //  这里是需要去发号机申请新的号段，如果使用同步请求，会降低性能


                LOGGER.info("开始尝试进行本地id段的同步操作");

                threadPoolExecutor.execute(() -> {
                    try {
                        IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                        tryUpdateMysqlRecord(idGeneratePO);
                    } catch (Exception e) {
                        LOGGER.error("同步id段失败,id is {}", localUnSeqIdBO.getId());
                    } finally {
                        semaphoreMap.get(localUnSeqIdBO.getId()).release();
                        LOGGER.info("本地无序id段同步完成,id is {}", localUnSeqIdBO.getId());
                    }

                });

            }
        }


    }

    //    bean初始化时会回调到这
    @Override
    public void afterPropertiesSet() throws Exception {

        List<IdGeneratePO> idGeneratePOList = idGenerateMapper.selectAll();

        for (IdGeneratePO idGeneratePO : idGeneratePOList) {
            tryUpdateMysqlRecord(idGeneratePO);
            semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));
        }
    }

    /**
     * 更新mysql里面的分布式id的配置信息，占用相应的id段
     * 同步执行，很多的网络IO，性能较慢
     *
     * @param idGeneratePO
     */
    private void tryUpdateMysqlRecord(IdGeneratePO idGeneratePO) {

        //告诉发号机 这个号段已经占用了，避免重复分配
        int updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());

        if (updateResult > 0) {
            //假设不存在并发问题，一切顺利，则将id号段缓存起来
            updateIdGenerateConfig(idGeneratePO);
            return; // 更新成功直接跳出
        }
        for (int i = 0; i < 3; i++) {
            //重复进行通知，直到发号机更新成功
            idGeneratePO = idGenerateMapper.selectById(idGeneratePO.getId());  // 重新获取新的version
            updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());

            if (updateResult > 0) {
                updateIdGenerateConfig(idGeneratePO);
                break;
            }
        }
        throw new RuntimeException("id生成机号段竞争失败");
    }


    private void updateIdGenerateConfig(IdGeneratePO idGeneratePO) {

        if (idGeneratePO.getIsSeq() == SEQ_ID) {
            //处理有序
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong currentStart = new AtomicLong(idGeneratePO.getCurrentStart());
            localSeqIdBO.setId(idGeneratePO.getId());
            localSeqIdBO.setCurrentNum(currentStart); //原子类
            localSeqIdBO.setCurrentStart(idGeneratePO.getCurrentStart());
            localSeqIdBO.setNextThreshold(idGeneratePO.getNextThreshold());
            localSeqIdBOMap.put(idGeneratePO.getId(), localSeqIdBO);
        }else {

            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGeneratePO.getId());

            //要实现把范围内的id加入队列，再把队列传入localUnSeqIdBO，队列的元素还需要打乱

            Long begin = idGeneratePO.getCurrentStart();
            Long end = idGeneratePO.getNextThreshold();

            ConcurrentLinkedQueue<Long> queue = new ConcurrentLinkedQueue<>();

//            for (long i = begin; i < end; i++) {
//                queue.add(i);
//            }
//            Collections.shuffle(queue);  //队列不能被打乱，换成 List

            List<Long> idlist = new ArrayList<>();

            for (long i = begin; i < end; i++) {
                idlist.add(i);
            }
            Collections.shuffle(idlist);
            queue.addAll(idlist);

            localUnSeqIdBO.setIdQueue(queue);


            localUnSeqIdBO.setCurrentStart(idGeneratePO.getCurrentStart());
            localUnSeqIdBO.setNextThreshold(idGeneratePO.getNextThreshold());
            localUnSeqIdBOMap.put(idGeneratePO.getId(), localUnSeqIdBO);
        }

    }
}




